package cn.tedu.rocket;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.junit.Test;

import java.nio.charset.StandardCharsets;

public class MyProducer {
    // 发送消息测试
    @Test
    public void send() throws Exception {
        //1. 创建具有连接rocketmq能力的生产者对象
        DefaultMQProducer producer = new DefaultMQProducer();
        // 提供给生产者nameserver地址
        producer.setNamesrvAddr("localhost:9876");
        // 每一个生产者都有自己的分组(名称自定义)
        producer.setProducerGroup("group01");
        // 启动连接
        producer.start();
        //2. 组织一个消息对象,封装业务数据
        Message message = new Message();
        // 发送的消息
        String msg = "我的第一条消息";
        byte[] bytes = msg.getBytes(StandardCharsets.UTF_8);
        // 消息体 - 字节数组
        message.setBody(bytes);
        // 绑定key值
        message.setKeys("order:sn:001");
        // 绑定tag值
        message.setTags("SZX");
        // 设置发送的目标(topic-主题)
        message.setTopic("topic03");

        //3. 调用api方法将消息发送 同步发送
        SendResult send = producer.send(message);
        //4. 接收发送结果 发送的状态 rockemtq中一些路由信息
        SendStatus sendStatus = send.getSendStatus();
        System.out.println("发送状态:"+sendStatus); //SEND_OK
        MessageQueue messageQueue = send.getMessageQueue();
        System.out.println("队列信息:"+messageQueue); //MessageQueue [topic=topic01, brokerName=DESKTOP-8CEMOMO, queueId=1]

        //5. 关闭生产者连接
        producer.shutdown();
    }

}







